package yuekao1.machine;

import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.dataproc.SplitBatchOp;
import com.alibaba.alink.operator.batch.evaluation.EvalRegressionBatchOp;
import com.alibaba.alink.operator.batch.source.CsvSourceBatchOp;
import com.alibaba.alink.operator.common.evaluation.RegressionMetrics;
import com.alibaba.alink.pipeline.dataproc.vector.VectorAssembler;
import com.alibaba.alink.pipeline.feature.OneHotEncoder;
import com.alibaba.alink.pipeline.regression.LassoRegression;
import com.alibaba.alink.pipeline.regression.RidgeRegression;

public class AbaloneMachine {
    public static void main(String[] args) throws Exception {
        //使用Flink批处理加载鲍鱼年龄数据集，封装数据集DataSet，提取特征features和标签label，前10条样本数据打印控制台；（3分）
        String filePath = "data/yk1/abalone.data";
        String schema
                //M,0.455,0.365,0.095,0.514,0.2245,0.101,0.15,15
                = "f0 String, f1 double, f2 double, f3 double, f4 double, f5 double, f6 double, f7 double, label int";
        CsvSourceBatchOp csvSource = new CsvSourceBatchOp()
                .setFilePath(filePath)
                .setSchemaStr(schema)
                .setFieldDelimiter(",");

//        csvSource.print();
        String[] features=new String[]{"f0","f1","f2","f3","f4","f5","f6","f7"};
        String label="label";

        OneHotEncoder one_hot = new OneHotEncoder()
                .setSelectedCols("f0")
                .setOutputCols("f00");
        BatchOperator<?> data = one_hot.fit(csvSource).transform(csvSource);
        VectorAssembler res = new VectorAssembler()
                .setSelectedCols("f00", "f1","f2","f3","f4","f5","f6","f7")
                .setOutputCol("vec");
        BatchOperator<?> data2 = res.transform(data);
        //（2）、按照8:2比例划分数据集：train训练集和test测试集，并查看条目数；（3分）
        BatchOperator<?> spliter = new SplitBatchOp().setFraction(0.8);
        BatchOperator<?> trainData = spliter.linkFrom(data2);
        BatchOperator<?> testData = spliter.getSideOutput(0);
        System.out.println("trainData条目数:"+trainData.count());
        System.out.println("testData条目数:"+testData.count());
        //（3）、调用 Alink 中 LassoRegression 回归API，应用train数据集训练构建模型，获取回归模型中截距intercept和参数θ值，并使用模型计算train、test数据集上MES值；（7分）
        //LassoRegression
        LassoRegression lasso = new LassoRegression()
                .setVectorCol("vec")
//                .setFeatureCols(features)
                .setLambda(0.1)
                .setLabelCol("label")
                .setPredictionCol("pred")
                .setMaxIter(1000)
                .enableLazyPrintModelInfo();
        BatchOperator<?> lasso_transform = lasso.fit(trainData).transform(testData);
        //（4）、对Alink中LassoRegression中各参数手动调整超参数，并在控制台输出每次调参时，模型在train、test数据集上MSE值；（5分）
        RegressionMetrics metrics = new EvalRegressionBatchOp()
                .setPredictionCol("pred")
                .setLabelCol("label")
                .linkFrom(lasso_transform)
                .collectMetrics();
        System.out.println("lassoMSE值:"+metrics.getMse());
        //（5）、调用Alink中 RidgeRegression 回归API，应用数据集训练构建模型，获取训练回归模型中截距intercept和参数θ值，并使用模型计算train、test数据集上MES值；（7分）
        RidgeRegression ridge = new RidgeRegression()
//                .setFeatureCols()
                .setVectorCol("vec")
                .setLambda(0.1)
                .setLabelCol("label")
                .setPredictionCol("pred")
                .setMaxIter(100)
                .enableLazyPrintModelInfo();
        BatchOperator<?> ridge_transform = ridge.fit(trainData).transform(testData);
        //（6）、对Alink中RidgeRegression中各参数手动调整超参数，并在控制台输出每次调参时，模型在train、test数据集上MSE值；（5分）
        RegressionMetrics metrics1 = new EvalRegressionBatchOp()
                .setPredictionCol("pred")
                .setLabelCol("label")
                .linkFrom(ridge_transform)
                .collectMetrics();
        System.out.println("ridgeMSE值:"+metrics1.getMse());
    }
}
